'''
@create_time: 2025/10/23 下午4:52
@Author: GeChao
@File: generate_test_report.py
'''

import argparse
from pathlib import Path
import os
from tools.db_connection import DbConnectionFactory
from tools.dingtalk_robot import DingTalkRobot
from tools.general_tool import sendMail
from tools.notification_manager import NotificationConfig
from tools.upload_to_cos import upload_to_cos
from tools.path_manage import ProjectFilePath
from datetime import datetime
import subprocess
import platform
import re

PROJECT_FILE_PATH = ProjectFilePath()
config = NotificationConfig()


def check_and_close_port(port: int):
    system_type = platform.system()

    # Windows 平台，使用 netstat + taskkill
    if system_type == "Windows":
        # 1. 查找占用 port 的进程信息
        cmd_find = f'netstat -ano | findstr :{port}'
        result = subprocess.run(cmd_find, shell=True, capture_output=True, text=True)

        if result.returncode == 0 and result.stdout.strip():
            lines = result.stdout.strip().split('\n')
            pids = set()

            # 2. 提取占用端口的 PID
            for line in lines:
                tokens = line.split()
                # 行格式一般为 TCP  0.0.0.0:80  0.0.0.0:0  LISTENING   1234(PID)
                pid = tokens[-1]  # 最后一个就是 PID
                pids.add(pid)

                # 3. 杀掉所有占用端口的进程
            for pid in pids:
                kill_cmd = f'taskkill /F /PID {pid}'
                subprocess.run(kill_cmd, shell=True)

            print(f"端口 {port} 被占用，已终止相关进程。")
        else:
            print(f"端口 {port} 未被占用。")
            # 类 Unix 系统(Linux、macOS)平台，使用 lsof + kill
    else:
        # 1. 根据端口查找占用端口的进程 PID
        cmd_find = ["lsof", "-i", f":{port}"]
        result = subprocess.run(cmd_find, capture_output=True, text=True)

        if result.returncode == 0 and result.stdout.strip():
            # 过滤出 PID 列
            lines = result.stdout.strip().split('\n')
            # 第一行是表头，如 "COMMAND   PID    USER   FD   TYPE  DEVICE SIZE/OFF NODE NAME"
            pids = set()
            for line in lines[1:]:
                tokens = re.split(r"\s+", line)
                if len(tokens) > 1:
                    pids.add(tokens[1])
                    # 2. 杀掉所有占用端口的进程
            for pid in pids:
                kill_cmd = ["kill", "-9", pid]
                subprocess.run(kill_cmd)
            print(f"端口 {port} 被占用，已终止相关进程。")
        else:
            print(f"端口 {port} 未被占用。")


def extract_cos_file_name(url):
    # 从URL中提取特定部分作为COS文件名。
    index = url.find("/auto_test/")
    return url[index:] if index != -1 else ''


def send_report_to_email(to_email: str):
    """
    发送测试报告到邮箱
    Args:
        to_email: 收件人邮箱地址
    """
    # 使用局域网IP地址,让同事也能访问报告
    # 请改为您电脑的局域网IP地址
    report_url = config.report_url  # 例如: http://192.168.1.100:38495
    current_date = datetime.now().strftime('%Y-%m-%d')

    project_file_path = ProjectFilePath()
    with DbConnectionFactory() as db_factory:
        test_result_db_path = getattr(project_file_path, 'test_result_db_path', '')
        sqlite_conn = db_factory.create_connection(
            connection_type='sqlite',
            db_path=test_result_db_path
        ).connect()

        # 查询统计数据
        exec_total_result = sqlite_conn.select(table='test_result_tb', fields=["count(*)"])
        exec_total = exec_total_result[0][0] if isinstance(exec_total_result, list) and len(exec_total_result) > 0 else 0
        
        exec_pass_result = sqlite_conn.select(
            table='test_result_tb',
            fields=["count(1)"],
            where={"execution_result": "pass"}
        )
        exec_pass_total = exec_pass_result[0][0] if isinstance(exec_pass_result, list) and len(exec_pass_result) > 0 else 0
        
        exec_fail_result = sqlite_conn.select(
            table='test_result_tb',
            fields=["count(*)"],
            where={"execution_result": "fail"}
        )
        exec_fail_total = exec_fail_result[0][0] if isinstance(exec_fail_result, list) and len(exec_fail_result) > 0 else 0
        
        exec_skip_result = sqlite_conn.select(
            table='test_result_tb',
            fields=["count(1)"],
            where={"execution_result": "skip"}
        )
        exec_skip_total = exec_skip_result[0][0] if isinstance(exec_skip_result, list) and len(exec_skip_result) > 0 else 0

        # 查询详细结果
        exec_pass_detail = sqlite_conn.execute(
            "select feature,count(1) from test_result_tb where execution_result='pass' group by feature;"
        )
        exec_fail_detail = sqlite_conn.execute(
            "select feature,title,count(1) from test_result_tb where execution_result='fail' group by title;"
        )

    # 计算通过率
    pass_rate = f"{(exec_pass_total / exec_total * 100):.1f}%" if exec_total > 0 else "0%"

    # 生成邮件正文(纯文本格式)
    email_content = f"""
自动化测试报告 ({current_date})
{'=' * 60}
测试概览:
- 总用例数: {exec_total} 条
- 通过: {exec_pass_total} 条
- 失败: {exec_fail_total} 条
- 跳过: {exec_skip_total} 条
- 通过率: {pass_rate}
{'-' * 60}
通过用例详情:
"""
    # 添加通过用例详情
    if exec_pass_detail:
        for i, item in enumerate(exec_pass_detail, 1):
            email_content += f"  {i}. {item[0]}: {item[1]} 条\n"
    else:
        email_content += "  无\n"
    email_content += f"\n{'-' * 60}\n\n失败用例详情:\n"
    # 添加失败用例详情
    if exec_fail_detail:
        for i, item in enumerate(exec_fail_detail, 1):
            email_content += f"  {i}. {item[0]} - {item[1]}: {item[2]} 条\n"
    else:
        email_content += "  无\n"

    email_content += f"\n{'-' * 60}\n\n"
    email_content += f"在线报告: {report_url}\n"
    email_content += f"\n注: 请确保 Allure 服务已启动，才能访问在线报告\n"

    # 发送邮件
    email_title = f"自动化测试报告 - {current_date}"
    success = sendMail(to_email, email_title, email_content)
    if success:
        print(f"✅ 邮件已成功发送到: {to_email}")
    else:
        print(f"❌ 邮件发送失败,请检查SMTP配置")


def send_report_to_dingtalk(webhook: str, secret: str = ''):
    """
    发送测试报告到钉钉群
    Args:
        webhook: 钉钉机器人的Webhook地址
        secret: 加签密钥(可选)
    """
    robot = DingTalkRobot(webhook, secret)
    # 使用局域网IP地址,让同事也能访问报告
    # 请改为您电脑的局域网IP地址
    report_url = config.report_url  # 例如: http://192.168.1.100:38495
    current_date = datetime.now().strftime('%Y-%m-%d')

    project_file_path = ProjectFilePath()
    test_result_db_path = getattr(project_file_path, 'test_result_db_path', '')
    with DbConnectionFactory() as db_factory:
        sqlite_conn = db_factory.create_connection(
            connection_type='sqlite',
            db_path=test_result_db_path
        ).connect()

        # 查询统计数据
        exec_total_result = sqlite_conn.select(table='test_result_tb', fields=["count(*)"])
        exec_total = exec_total_result[0][0] if isinstance(exec_total_result, list) and len(exec_total_result) > 0 else 0
        
        exec_pass_result = sqlite_conn.select(
            table='test_result_tb',
            fields=["count(1)"],
            where={"execution_result": "pass"}
        )
        exec_pass_total = exec_pass_result[0][0] if isinstance(exec_pass_result, list) and len(exec_pass_result) > 0 else 0
        
        exec_fail_result = sqlite_conn.select(
            table='test_result_tb',
            fields=["count(*)"],
            where={"execution_result": "fail"}
        )
        exec_fail_total = exec_fail_result[0][0] if isinstance(exec_fail_result, list) and len(exec_fail_result) > 0 else 0
        
        exec_skip_result = sqlite_conn.select(
            table='test_result_tb',
            fields=["count(1)"],
            where={"execution_result": "skip"}
        )
        exec_skip_total = exec_skip_result[0][0] if isinstance(exec_skip_result, list) and len(exec_skip_result) > 0 else 0

        # 查询详细结果
        exec_pass_detail = sqlite_conn.execute(
            "select feature,count(1) from test_result_tb where execution_result='pass' group by feature;"
        )
        exec_fail_detail = sqlite_conn.execute(
            "select feature,title,count(1),picture,picture_name from test_result_tb where execution_result='fail' group by title;"
        )

        # 生成钉钉Markdown格式详细信息
        pass_detail = generate_dingtalk_detail(exec_pass_detail, 'pass')
        fail_detail = generate_dingtalk_detail(exec_fail_detail, 'fail')

        # 构造钉钉Markdown消息
        pass_rate = f"{(exec_pass_total / exec_total * 100):.1f}%" if exec_total > 0 else "0%"

        markdown_text = f"""
### 📊 自动化测试报告 ({current_date})
---
#### 测试概览
- **总用例数**: {exec_total} 条
- **通过**: <font color=#67C23A>{exec_pass_total}</font> 条
- **失败**: <font color=#F56C6C>{exec_fail_total}</font> 条  
- **跳过**: <font color=#909399>{exec_skip_total}</font> 条
- **通过率**: <font color=#409EFF>{pass_rate}</font>
---
{pass_detail}
---
{fail_detail}
---
#### 🔗 在线报告
点击查看详细报告: [{report_url}]({report_url})
        """.strip()

        # 释放端口并发送消息
        check_and_close_port(38495)
        robot.send_markdown(
            title=f"自动化测试报告-{current_date}",
            text=markdown_text
        )


def generate_dingtalk_detail(exec_result, result_type):
    """
    生成钉钉Markdown格式的详细信息
    
    Args:
        exec_result: 执行结果
        result_type: 结果类型 (pass/fail)
        
    Returns:
        Markdown格式的字符串
    """
    if not exec_result:
        return ''

    if result_type == 'pass':
        title = "#### ✅ 通过用例详情"
        details = []
        for i, item in enumerate(exec_result, 1):
            details.append(f"- **{item[0]}**: {item[1]} 条")
        newline = '\n'
        return f"{title}\n{newline.join(details)}"

    else:  # fail
        title = "#### ❌ 失败用例详情"
        details = []
        for i, item in enumerate(exec_result, 1):
            detail_line = f"- **{item[0]} - {item[1]}**: {item[2]} 条"
            # 如果有截图链接
            if len(item) > 4 and item[4]:
                detail_line += f" [📸查看截图]({item[4]})"
            details.append(detail_line)
        newline = '\n'
        return f"{title}\n{newline.join(details)}"


if __name__ == '__main__':
    parser = argparse.ArgumentParser(prog='查看报告命令行')
    parser.add_argument('-a', '--allReport', action='store_true', help='查看总测试报告')
    parser.add_argument('-wpt', '--webReport', action='store_true', help='查看web测试报告')
    parser.add_argument('-g', '--generateOnly', action='store_true', default=True, help='默认生成静态报告')
    parser.add_argument('--no-generateOnly', action='store_false', dest='generateOnly', help='启动Web服务器而不是生成静态报告')

    # 钉钉相关参数
    parser.add_argument('-dt', '--dingtalk', default='0', help='是否发送到钉钉群：0.否，1.是 (默认 0)', type=str)
    parser.add_argument('-dtw', '--dingtalkWebhook', type=str, help='钉钉机器人Webhook地址')
    parser.add_argument('-dts', '--dingtalkSecret', type=str, help='钉钉机器人加签密钥(可选)')

    # 邮件相关参数
    parser.add_argument('-em', '--email', default='0', help='是否发送邮件：0.否，1.是 (默认 0)', type=str)
    parser.add_argument('-emt', '--emailTo', type=str, help='收件人邮箱地址')

    args = parser.parse_args()

    # 钉钉机器人
    if args.dingtalk == '1':
        if not args.dingtalkWebhook:
            print("❌ 错误: 需要提供钉钉Webhook地址 (--dingtalkWebhook)")
        else:
            send_report_to_dingtalk(
                webhook=args.dingtalkWebhook,
                secret=args.dingtalkSecret or ''
            )

    # 邮件通知
    if args.email == '1':
        if not args.emailTo:
            print("❌ 错误: 需要提供收件人邮箱地址 (--emailTo)")
        else:
            send_report_to_email(to_email=args.emailTo)

    # 启动Allure报告服务
    if args.allReport:
        project_file_path = ProjectFilePath()
        test_case_report_path = getattr(project_file_path, 'test_case_report_path', Path('.'))
        # 默认生成静态报告，除非明确指定要启动服务器
        if args.generateOnly:
            # 仅生成静态报告，不启动服务器
            os.system(
                f'allure generate {str(test_case_report_path / "allure-results")} -o {str(test_case_report_path / "allure-report")} --clean')
            print(f"✅ 静态报告已生成到: {str(test_case_report_path / 'allure-report')}")
        else:
            # 启动Web服务器
            os.system(
                f'allure serve {str(test_case_report_path / "allure-results")} -p 38495 --host 0.0.0.0')
